package com.qushiyun.devops.admin.sync.job;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.Wrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.qushiyun.devops.admin.sync.data.mapper.TableLogMapper;
import com.qushiyun.devops.admin.sync.data.po.TableLog;
import com.qushiyun.devops.admin.sync.data.po.TriggerJob;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import lombok.extern.slf4j.Slf4j;
import org.anyline.data.jdbc.ds.DataSourceHolder;
import org.anyline.data.jdbc.runtime.JDBCRuntimeHolder;
import org.anyline.data.param.ConfigStore;
import org.anyline.data.param.init.DefaultConfigStore;
import org.anyline.entity.Compare;
import org.anyline.entity.DataRow;
import org.anyline.entity.DataSet;
import org.anyline.entity.Order.TYPE;
import org.anyline.metadata.Column;
import org.anyline.metadata.Table;
import org.anyline.proxy.CacheProxy;
import org.anyline.service.AnylineService;
import org.anyline.util.BasicUtil;
import org.springframework.stereotype.Component;

/**
 * 同步任务
 *
 * @author zhouzx
 */
@Slf4j
@Component
public class SyncTask {

    private final AnylineService service;

    private final TableLogMapper tableLogMapper;
    private final FunctionHelper functionHelper;

    private static final String ADD_TENANT_ID = "ALTER TABLE %s ADD COLUMN `tenant_id` varchar(32) NULL COMMENT '租户Id' AFTER  %s";

    public SyncTask(AnylineService service, TableLogMapper tableLogMapper, FunctionHelper functionHelper) {
        this.service = service;
        this.tableLogMapper = tableLogMapper;
        this.functionHelper = functionHelper;
    }


    public synchronized void execute(long id) throws Exception {
        DataSourceHolder.setDefaultDataSource();
        ConfigStore conditions = new DefaultConfigStore();
        conditions.and("id", id);
        DataSet tasks = service.querys("sync_trigger_job", conditions);
        int syncType = tasks.getInt("trigger_type");
        if (syncType == 3) {
            conditions = new DefaultConfigStore();
            conditions.and("id", tasks.get(0, "job_ids"));
            DataSet set = service.querys("sync_trigger_job", conditions);
            for (int i = 0; i < set.size(); i++) {
                DataRow task = tasks.getRow(i);
                taskRun(task);
            }
        }
        taskRun(tasks.getRow(0));
    }

    private void taskRun(DataRow task) throws Exception {
        String fromDb = "from_db", toDb = "to_db";
        try {
            long start = System.currentTimeMillis();
            log.info("{} 任务开始执行", task.getString("name"));
            String tenantId = task.getString("tenant_id");

            //注册datasource
            fromDb = regDb(task, fromDb);
            toDb = regDb(task, toDb);

            TriggerJob job = new TriggerJob();
            job.setTriggerTime(new Date());
            job.setId(task.getLong("id"));
            int status = 2;

            int conditionId = task.getInt("condition_id", 0);
            String fromBeforeSql = task.getString("from_db_before_sql");
            String toBeforeSql = task.getString("to_db_before_sql");
            DataSourceHolder.setDataSource(fromDb);
            if (StrUtil.isNotBlank(fromBeforeSql)) {
                service.execute(fromBeforeSql);
            }
            DataSourceHolder.setDataSource(toDb);
            if (StrUtil.isNotBlank(toBeforeSql)) {
                service.execute(toBeforeSql);
            }
            DataSourceHolder.setDefaultDataSource();
            DataSet condition = service.querys("sync_condition_config", "id:" + conditionId);
            DataRow conRow = condition.getRow(0);
            exe(job, fromDb, toDb, conRow, tenantId);

            String fromDbAfterSql = task.getString("from_db_after_sql");
            String toDbAfterSql = task.getString("to_db_after_sql");
            DataSourceHolder.setDataSource(fromDb);
            if (StrUtil.isNotBlank(fromDbAfterSql)) {
                service.execute(fromDbAfterSql);
            }
            DataSourceHolder.setDataSource(toDb);
            if (StrUtil.isNotBlank(toDbAfterSql)) {
                service.execute(toDbAfterSql);
            }

            long costTime = System.currentTimeMillis() - start;
            DataSourceHolder.setDefaultDataSource();
            task.put("cost_time", costTime);
            task.put("status", status);
            task.put("data_num", job.getDataNum());
            task.put("table_num", job.getTableNum());
            task.put("trigger_time", job.getTriggerTime());
            service.update(task, "status", "cost_time", "data_num", "table_num", "trigger_time");
            log.info("{} 任务执行完成，耗时：{}", task.getString("name"), costTime);
        } catch (Exception e) {
            String fromDbErrorSql = task.getString("from_db_error_sql");
            String toDbErrorSql = task.getString("to_db_error_sql");
            DataSourceHolder.setDataSource(fromDb);
            if (StrUtil.isNotBlank(fromDbErrorSql)) {
                service.execute(fromDbErrorSql);
            }
            DataSourceHolder.setDataSource(toDb);
            if (StrUtil.isNotBlank(toDbErrorSql)) {
                service.execute(toDbErrorSql);
            }
            throw e;
        }
    }

    private String regDb(DataRow task, String dbId) {
        DataSet from = service.querys("sync_data_source", "id:" + task.get(dbId));
        for (DataRow ds : from) {
            String code = ds.getString("id");
            dbId = code;
            String driver = ds.getString("driver");
            String url = ds.getString("url");
            String account = ds.getString("username");
            String password = ds.getString("password");
            String pool = "com.zaxxer.hikari.HikariDataSource";
            try {
                if (!DataSourceHolder.contains(code)) {
                    DataSourceHolder.reg(code, pool, driver, url, account, password);
                } else {
                    JDBCRuntimeHolder.destroy(code);
                    DataSourceHolder.reg(code, pool, driver, url, account, password);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        return dbId;
    }


    /**
     * 执行同步(0:全量  1:增量  2:更新)
     */
    private TriggerJob exe(TriggerJob job, String fromDb, String toDb, DataRow task, String tenantId) throws Exception {
        job.setTriggerTime(new Date());
        DataSourceHolder.setDefaultDataSource();
        List<String> tables = (List<String>) task.get("table_list");
        Collection<String> ignores = (List) task.get("ignore_table");
        Collection<String> noDataTable = (Collection) task.get("no_data_table");
        long triggerId = job.getId();
        Wrapper<TableLog> wrapper = Wrappers.lambdaQuery(TableLog.class).eq(TableLog::getTriggerId, triggerId);
        tableLogMapper.delete(wrapper);

        int overWrite = task.getInt("over_write");

        if (CollUtil.isEmpty(tables)) {
            log.error("未设置源表:{}", task.toJSON());
            return job;
        }
        DataSourceHolder.setDataSource(fromDb);
        if (tables.size() == 1 && tables.get(0).equals("*")) {
            CacheProxy.clear();
            tables = service.tables();
        } else {
            CacheProxy.clear();
            List<String> dbTables = service.tables();
            List<String> newTables = new ArrayList<>();
            for (String dbTable : dbTables) {
                for (String table : tables) {
                    if (dbTable.contains(table)) {
                        newTables.add(dbTable);
                    }
                }
            }
            tables = newTables;
        }

        functionHelper.sync(fromDb, toDb);

        long index = 0, dataNum = 0;
        int tableNum = 0;
        for (String table : tables) {
            long start = System.currentTimeMillis();
            index++;
            log.info("开始同步第 {} 张表 {}", index, table);
            TableLog tableLog = new TableLog();
            tableLog.setBeginTime(new Date());
            boolean ignored = false;
            if (CollUtil.isNotEmpty(ignores)) {
                for (String ignore : ignores) {
                    if (table.contains(ignore)) {
                        ignored = true;
                        break;
                    }
                }
                if (ignored) {
                    tableLog.setIsIgnore(1);
                    tableLog.setEndTime(new Date());
                    tableLog.setCostTime((System.currentTimeMillis() - start));
                    tableLogMapper.insert(tableLog);
                    continue;
                }
            }
            DataSourceHolder.setDataSource(fromDb);
            List<String> ddl = service.metadata().ddl(table);
            if (ddl == null || ddl.size() == 0) {
                log.warn("{} table ddl is empty ", table);
                continue;
            }
            tableNum++;
            DataSourceHolder.setDataSource(toDb);
            Table t = new Table(table);

            boolean exists = service.metadata().exists(t);

            if (!exists || (exists && overWrite == 1)) {
                service.execute("drop table if exists " + table + ";");
                service.execute(ddl.get(0));
            }
            ignored = false;
            if (CollUtil.isNotEmpty(noDataTable)) {
                for (String ignore : noDataTable) {
                    if (table.contains(ignore)) {
                        ignored = true;
                        break;
                    }
                }
                if (ignored) {
                    log.warn("{}只同步表结构，不同步数据", table);
                    tableLog.setDataNum(0L);
                    tableLog.setStatus(2);
                    tableLog.setIsCreated(1);
                    tableLog.setEndTime(new Date());
                    tableLog.setCostTime(System.currentTimeMillis() - start);
                    tableLogMapper.insert(tableLog);
                    continue;
                }
            }

            DataSourceHolder.setDataSource(fromDb);
            task.put("trigger_id", job.getId());
            task.put("SRC_TABLE", table);
            task.put("TAR_TABLE", table);
            task.put("tenant_id", tenantId);
            task.put("COLS", "*");
            long size = exportAndImport(fromDb, toDb, task, 0);
            dataNum += size;
            log.info("同步表:{}，同步数据量：{}，耗时：{}", table, size, System.currentTimeMillis() - start);
            DataSourceHolder.setDataSource(fromDb);
        }
        job.setDataNum(dataNum);
        job.setTableNum(tableNum);
        return job;
    }

    private long exportAndImport(String fromDb, String toDb, DataRow task, int type) throws Exception {
        List<String> all = (List) task.get("all_table");
        List<String> tenantAndGlobal = (List) task.get("tenant_global_table");
        String conditionStr = task.getString("condition_str");
        int overWrite = task.getInt("over_write");
        int conditionType = task.getInt("type");
        String tenantId = task.getString("tenant_id");

        if (StrUtil.isBlank(tenantId)) {
            log.error("租户ID为空，直接返回");
            return 0;
        }
        List<String> tenants = new ArrayList<>();
        if (tenantId.contains(",")) {
            tenants = StrUtil.split(tenantId, ",");
        } else {
            tenants.add(tenantId);
        }
        int step = task.getInt("STEP", 200);
        String srcTable = task.getString("SRC_TABLE");
        String tarTable = srcTable;
        String orderKey = task.getStringNvl("ORDER_COLUMN");
        TableLog tableLog = new TableLog();
        tableLog.setBeginTime(new Date());
        tableLog.setTriggerId(task.getLong("trigger_id"));
        long start = System.currentTimeMillis();
        tableLog.setTableName(srcTable);
        tableLog.setIsCreated(1);
        if (BasicUtil.isEmpty(orderKey)) {
            DataSourceHolder.setDataSource(fromDb);
            Table table = service.metadata().table(srcTable);
            LinkedHashMap<String, Column> columns = table.getColumns();
            boolean hasTenantIdColumn = false;
            for (Entry<String, Column> entry : columns.entrySet()) {
                String k = entry.getKey();
                if (k.equalsIgnoreCase("tenant_id")) {
                    hasTenantIdColumn = true;
                    break;
                }
            }
            if (!hasTenantIdColumn) {
                log.error("{} 表没有租户Id", srcTable);
            }
            Column col = table.primary();
            if (null != col) {
                orderKey = col.getName();
                if (!hasTenantIdColumn) {
                    //service.execute(String.format(ADD_TENANT_ID, srcTable, orderKey));
                }
                if (type == 0) {
                    DataSourceHolder.setDataSource(fromDb);
                } else {
                    log.error("{} 表没有主键", srcTable);
                }
            }
            String max = "0";
            String[] cols = new String[0];
            //全量
            if (type == 0) {
                if (overWrite == 1) {
                    tableLog.setOverWrite(1);
                    DataSourceHolder.setDataSource(toDb);
                    service.execute("TRUNCATE TABLE " + tarTable);
                }
                DataSourceHolder.setDataSource(fromDb);

                boolean allSync = false;
                if (CollUtil.isNotEmpty(all)) {
                    for (String allTableName : all) {
                        if (srcTable.contains(allTableName)) {
                            allSync = true;
                            break;
                        }
                    }
                }

                if (allSync) {
                    tableLog.setIsAll(1);
                    DataSourceHolder.setDataSource(toDb);
                    service.execute("TRUNCATE TABLE " + tarTable);
                    DataSourceHolder.setDataSource(fromDb);
                    DataSet set = service.querys(srcTable);
                    if (set.size() != 0) {
                        set = format(set, cols);
                        DataSourceHolder.setDataSource(toDb);
                        service.insert(tarTable, set);
                    }

                    tableLog.setDataNum((long) set.size());
                    tableLog.setStatus(2);
                    tableLog.setEndTime(new Date());
                    tableLog.setCostTime(System.currentTimeMillis() - start);
                    tableLogMapper.insert(tableLog);
                    return set.size();
                }

                ConfigStore condition = new DefaultConfigStore(0, step - 1);
                if (conditionType == 1) {
                    if (CollUtil.isNotEmpty(tenantAndGlobal) && tenantAndGlobal.contains(srcTable)) {
                        tenants.add("-1");
                        condition.and("tenant_id", tenants);
                        tableLog.setIsTenantGlobal(1);
                    } else {
                        condition.and("tenant_id", tenants);
                    }
                } else if (conditionType == 2 && StrUtil.isNotBlank(conditionStr)) {
                    condition.and(conditionStr);
                }
                long exeQty = 0;
                while (true) {
                    DataSourceHolder.setDataSource(fromDb);
                    condition.and(Compare.GREAT, orderKey, max);
                    condition.order(orderKey, TYPE.ASC);
                    DataSet set = service.querys(srcTable, condition);
                    max = set.getString(set.size() - 1, orderKey, "0");
                    if (set.size() == 0) {
                        break;
                    }
                    exeQty += set.size();
                    set = format(set, cols);
                    DataSourceHolder.setDataSource(toDb);
                    service.insert(tarTable, set);
                }
                tableLog.setDataNum(exeQty);
                tableLog.setStatus(2);
                tableLog.setEndTime(new Date());
                tableLog.setCostTime(System.currentTimeMillis() - start);
                tableLogMapper.insert(tableLog);
                return exeQty;
            }
        }

        return 0;
    }

    private DataSet format(DataSet set, String... cols) {
        List<DataRow> rows = set.getRows();
        Iterator<DataRow> iterator = rows.iterator();
        while (iterator.hasNext()) {
            DataRow row = iterator.next();
            if (row != null) {
                Iterator<Map.Entry<String, Object>> d = row.entrySet().iterator();
                while (d.hasNext()) {
                    Map.Entry<String, Object> entry = d.next();
                    Object v = entry.getValue();
                    if (v instanceof List || v instanceof DataSet) {
                        if (((Collection<?>) v).size() == 0) {
                            row.put(entry.getKey(), "[]");
                        } else {
                            row.put(entry.getKey(), JSON.toJSONString(v));
                        }
                    }

                    if (v instanceof DataRow) {
                        if (((DataRow) v).size() == 0) {
                            row.put(entry.getKey(), "{}");
                        } else {
                            row.put(entry.getKey(), JSON.toJSONString(v));
                        }
                    }

                    if (v == null) {
                        row.put(entry.getKey(), null);
                    }
                }
            }
        }

        List<String> inserts = new ArrayList<>();
        if (null != cols && cols.length > 0) {
            for (String col : cols) {
                String[] tmps = col.split(":");
                if (tmps.length > 1) {
                    if (!tmps[0].equalsIgnoreCase(tmps[1])) {
                        set.changeKey(tmps[0], tmps[1], true);
                    }
                    inserts.add(tmps[1]);
                } else {
                    inserts.add(col);
                }
            }
        } else {
            return set;
        }

        return set.extract(inserts);
    }
}